package com.gz.rocketmq.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.platform.commons.util.CollectionUtils;
import org.junit.platform.commons.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 事务消息
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("rocketT");
        producer.setNamesrvAddr("192.168.36.128:9876");
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if("tagA".equals(message.getTags())){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if("tagB".equals(message.getTags())){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                if(messageExt.getTags().equals("tagC")){
                    System.out.println("tagC   检查本地事务状态");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });

        producer.start();
        String[] tags = new String[]{"tagA","tagB","tagC"};

        for (int i = 0; i < 3; i++) {
            Message message = new Message("tranTopic",tags[i],("事务消息"+i).getBytes(StandardCharsets.UTF_8));
            TransactionSendResult result = producer.sendMessageInTransaction(message, null);
            System.out.println(result);
        }


    }
}
